package org.apache.rocketmq.proxy.processor;

import io.netty.channel.Channel;
import org.apache.rocketmq.broker.client.ClientChannelInfo;
import org.apache.rocketmq.broker.client.ConsumerGroupInfo;
import org.apache.rocketmq.broker.client.ConsumerIdsChangeListener;
import org.apache.rocketmq.broker.client.ProducerChangeListener;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.proxy.common.ProxyContext;
import org.apache.rocketmq.proxy.service.ServiceManager;
import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;


import java.util.Set;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/13
 * @方法描述：客户端处理器
 */
public class ClientProcessor extends AbstractProcessor {

    public ClientProcessor(MessagingProcessor messagingProcessor,
                           ServiceManager serviceManager) {
        super(messagingProcessor, serviceManager);
    }

    public void registerProducer(
            ProxyContext ctx,
            String producerGroup,
            ClientChannelInfo clientChannelInfo
    ) {
        this.serviceManager.getProducerManager().registerProducer(producerGroup, clientChannelInfo);
    }

    public void unRegisterProducer(
            ProxyContext ctx,
            String producerGroup,
            ClientChannelInfo clientChannelInfo
    ) {
        this.serviceManager.getProducerManager().unregisterProducer(producerGroup, clientChannelInfo);
    }

    public Channel findProducerChannel(
            ProxyContext ctx,
            String producerGroup,
            String clientId
    ) {
        return this.serviceManager.getProducerManager().findChannel(clientId);
    }

    public void registerProducerChangeListener(ProducerChangeListener listener) {
        this.serviceManager.getProducerManager().appendProducerChangeListener(listener);
    }

    public void registerConsumer(
            ProxyContext ctx,
            String consumerGroup,
            ClientChannelInfo clientChannelInfo,
            ConsumeType consumeType,
            MessageModel messageModel,
            ConsumeFromWhere consumeFromWhere,
            Set<SubscriptionData> subList,
            boolean updateSubscription
    ) {
        this.serviceManager.getConsumerManager().registerConsumer(
                consumerGroup,
                clientChannelInfo,
                consumeType,
                messageModel,
                consumeFromWhere,
                subList,
                false,
                updateSubscription);
    }

    public ClientChannelInfo findConsumerChannel(
            ProxyContext ctx,
            String consumerGroup,
            Channel channel
    ) {
        return this.serviceManager.getConsumerManager().findChannel(consumerGroup, channel);
    }

    public void unRegisterConsumer(
            ProxyContext ctx,
            String consumerGroup,
            ClientChannelInfo clientChannelInfo
    ) {
        this.serviceManager.getConsumerManager().unregisterConsumer(consumerGroup, clientChannelInfo, false);
    }

    public void doChannelCloseEvent(String remoteAddr, Channel channel) {
        this.serviceManager.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
        this.serviceManager.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
    }

    public void registerConsumerIdsChangeListener(ConsumerIdsChangeListener listener) {
        this.serviceManager.getConsumerManager().appendConsumerIdsChangeListener(listener);
    }

    public ConsumerGroupInfo getConsumerGroupInfo(ProxyContext ctx, String consumerGroup) {
        return this.serviceManager.getConsumerManager().getConsumerGroupInfo(consumerGroup);
    }
}

